package org.example.infrastructure.repository.orm.messageQueue.consumer;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.example.domain.model.entity.url.*;
import org.example.domain.model.entity.url.vo.LocationInfo;
import org.example.domain.model.entity.url.vo.UrlShortenerStatsRecordVo;
import org.example.domain.repository.*;
import org.example.infrastructure.config.AmapRemoteConfig;
import org.example.infrastructure.repository.orm.messageQueue.constant.UrlShortenerCoreMQConstant;
import org.example.infrastructure.repository.orm.messageQueue.idempotent.MessageQueueIdempotentHandler;
import org.example.infrastructure.repository.orm.redis.config.constant.RedisCacheConstant;
import org.example.web.exception.ServerException;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = UrlShortenerCoreMQConstant.STATA_SAVE_TOPIC,
        consumerGroup = "${rocketmq.consumer.group}"
)
public class UrlShortenerCoreStatsSaveConsumer implements RocketMQListener<Map<String, String>> {
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    private final RedissonClient redissonClient;
    private final UrlShortenerRepository urlShortenerRepository;
    private final UrlShortenerCorrelationGroupMapperRepository urlShortenerCorrelationGroupMapperRepository;
    private final UrlShortenerAccessStatsRepository urlShortenerAccessStatsRepository;
    private final UrlShortenerLocaleStatsRepository urlShortenerLocaleStatsRepository;
    private final UrlShortenerOsStatsRepository urlShortenerOsStatsRepository;
    private final UrlShortenerBrowserStatsRepository urlShortenerBrowserStatsRepository;
    private final UrlShortenerDeviceStatsRepository urlShortenerDeviceStatsRepository;
    private final UrlShortenerNetworkStatsRepository urlShortenerNetworkStatsRepository;
    private final UrlShortenerAccessLogsRepository urlShortenerAccessLogsRepository;
    private final UrlShortenerStatsTodayRepository urlShortenerStatsTodayRepository;
    private final AmapRemoteConfig amapRemoteConfig;



    @Override
    public void onMessage(Map<String, String> producerMap) {
        String keys = producerMap.get(MessageConst.PROPERTY_KEYS);
        if (messageQueueIdempotentHandler.isMessageBeingConsumed(keys)){
            //判断当前这个消息流程是否执行完成
            if (messageQueueIdempotentHandler.isAccomplish(keys)){
                return;
            }
            throw new ServerException("消息未完成流程，需要消息队列重试");
        }
        try {
            UrlShortenerStatsRecordVo statsRecord = JSON.parseObject(producerMap.get("statsRecord"), UrlShortenerStatsRecordVo.class);
            actualSaveShortLinkStats(statsRecord);
        } catch (Throwable e) {
            log.error("短链接监控消费异常", e);
            try {
                messageQueueIdempotentHandler.delMessageProcessed(keys);
            }catch (Throwable remoteEx){
                log.error("删除幂等标识错误", remoteEx);
            }
        }

        messageQueueIdempotentHandler.setAccomplish(keys);

    }

    public void actualSaveShortLinkStats(UrlShortenerStatsRecordVo statsRecord) {
        String fullShortUrl = statsRecord.getFullShortUrl();
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisCacheConstant.LOCK_GID_UPDATE_KEY + fullShortUrl);
        RLock rLock = readWriteLock.readLock();
        rLock.lock();
        try {
            UrlShortenerCorrelationGroup urlShortenerCorrelationGroup = urlShortenerCorrelationGroupMapperRepository.selectOneByFullShortUrl(fullShortUrl);
            String gid = urlShortenerCorrelationGroup.getGid();
            Date currentDate = statsRecord.getCurrentDate();
            int hour = DateUtil.hour(currentDate, true);
            Week week = DateUtil.dayOfWeekEnum(currentDate);
            int weekValue = week.getIso8601Value();
            UrlShortenerAccessStats urlShortenerAccessStats = UrlShortenerAccessStats.builder()
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .hour(hour)
                    .weekday(weekValue)
                    .fullShortUrl(fullShortUrl)
                    .accessDate(currentDate)
                    .build();
            urlShortenerAccessStatsRepository.insertUrlShortenerAccessStats(urlShortenerAccessStats);
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", amapRemoteConfig.getLocaleAmapKey());
            localeParamMap.put("ip", statsRecord.getRemoteAddr());
            String localeResultStr = HttpUtil.get(amapRemoteConfig.getAmapIpRemoteUrl(), localeParamMap);
            LocationInfo locationInfo = JSON.parseObject(localeResultStr, LocationInfo.class);
            final String infoCode = locationInfo.getInfocode();
            final String defaultValue = "未知";
            if (CharSequenceUtil.isNotBlank(infoCode) && CharSequenceUtil.equals("10000", infoCode)) {
                final String province = locationInfo.getProvince();
                boolean unknownFlag = CharSequenceUtil.equals(province, "[]");
                UrlShortenerLocaleStats urlShortenerLocaleStats = UrlShortenerLocaleStats.builder()
                        .province(unknownFlag ? defaultValue : province)
                        .city(unknownFlag ? defaultValue : locationInfo.getCity())
                        .adcode(unknownFlag ? defaultValue : locationInfo.getAdcode())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .country("中国")
                        .accessDate(currentDate)
                        .build();
                urlShortenerLocaleStatsRepository.insertUrlShortenerLocaleStats(urlShortenerLocaleStats);
            }
            UrlShortenerOsStats urlShortenerOsStats = UrlShortenerOsStats.builder()
                    .os(statsRecord.getOs())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .accessDate(currentDate)
                    .build();
            urlShortenerOsStatsRepository.insertUrlShortenerOsStats(urlShortenerOsStats);
            UrlShortenerBrowserStats urlShortenerBrowserStats = UrlShortenerBrowserStats.builder()
                    .browser(statsRecord.getBrowser())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .accessDate(currentDate)
                    .build();
            urlShortenerBrowserStatsRepository.insertUrlShortenerBrowserStats(urlShortenerBrowserStats);

            UrlShortenerDeviceStats urlShortenerDeviceStats = UrlShortenerDeviceStats.builder()
                    .device(statsRecord.getDevice())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .accessDate(currentDate)
                    .build();
            urlShortenerDeviceStatsRepository.insertUrlShortenerDeviceStats(urlShortenerDeviceStats);

            UrlShortenerNetworkStats urlShortenerNetworkStats = UrlShortenerNetworkStats.builder()
                    .network(statsRecord.getNetwork())
                    .cnt(1)
                    .fullShortUrl(fullShortUrl)
                    .accessDate(currentDate)
                    .build();
            urlShortenerNetworkStatsRepository.insertUrlShortenerNetworkStats(urlShortenerNetworkStats);


            UrlShortenerAccessLogs urlShortenerAccessLogs = UrlShortenerAccessLogs.builder()
                    .uv(statsRecord.getUv())
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .network(statsRecord.getNetwork())
                    .device(statsRecord.getDevice())
                    .locale(CharSequenceUtil.join("-", "中国", locationInfo.getProvince(), locationInfo.getCity()))
                    .fullShortUrl(fullShortUrl)
                    .build();
            urlShortenerAccessLogsRepository.insertUrlShortenerAccessLogs(urlShortenerAccessLogs);

            urlShortenerRepository.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);

            UrlShortenerStatsToday urlShortenerStatsToday = UrlShortenerStatsToday.builder()
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .fullShortUrl(fullShortUrl)
                    .accessDate(currentDate)
                    .build();
            urlShortenerStatsTodayRepository.insertUrlShortenerStatsToday(urlShortenerStatsToday);
        } catch (Throwable ex) {
            log.error("短链接访问量统计异常", ex);
        } finally {
            rLock.unlock();
        }
    }
}
